package com.spi.listener;


import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.spi.feign.User;
import com.spi.feign.UserFeign;
import com.spi.mapper.SurveyMapper;
import com.spi.model.Survey;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.HashMap;

@RocketMQMessageListener(topic = "${rocketmq.consumer.group2.topic}",
        consumerGroup = "${rocketmq.consumer.group2.consumer-group}",
        maxReconsumeTimes = 3)
@Component
@Slf4j
public class Topic2Listener implements RocketMQListener<String> {

    @Autowired
    private UserFeign userFeign ;

    @Autowired
    private SurveyMapper surveyMapper ;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void onMessage(String s) {
        String messageid = null ;
        try {
            HashMap hashMap = JSONObject.parseObject(s, HashMap.class);
            User user = JSONObject.parseObject((String) hashMap.get("user"), User.class);
            messageid = (String)hashMap.get("messageid"); //消息id
            //1:消费者端幂等性验证
            int cousumerStatus = userFeign.getCousumerStatus(messageid);
            if (cousumerStatus == 1) {
                log.info("该条消息已经消费过。。 直接返回");
                return;
            }
            //2：消费者端执行本地事务
            Survey survey = new Survey();
            survey.setUserid("");
            LambdaQueryWrapper<Survey> lambdaQueryWrapper = new LambdaQueryWrapper<>();
            lambdaQueryWrapper.eq(Survey::getUserid, user.getUserid());
            surveyMapper.update(survey,lambdaQueryWrapper);
            //3：记录事务消息表消费者消费状态
            int i = userFeign.updateCousumerStatusByMessageId(messageid,1);
            if (i == 1){
                log.info("消费者端修改事务消息表消费者消费状态成功。。。。");
            }else {
                log.info("消费者端修改事务消息表消费者消费状态失败。。。。");
            }
            //throw new RuntimeException("手动触发异常。。。");

        }catch (Exception e){
            userFeign.updateCousumerStatusByMessageId(messageid,2);
            e.printStackTrace();
            throw new RuntimeException("消费者消费异常。。" + e.getMessage());
        }
    }
}
